package com.hu.flink12.api.transformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: hujianjun
 * @Date: 2021/2/4 0:17
 * @Describe: 模拟数据倾斜，解决方法是算子调用rebanlance重分布，原理是内部使用round robin方法将数据均匀打散
 *
 *   统计分到不同子任务/分区上的数据量多少观察是否产生了数据倾斜
 *
 */
public class RebalanceStream {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //TODO 2.读取数据源
        DataStream<Long> ds = env.fromSequence(1, 100);

        //TODO 3.转换操作
        DataStream<Long> filterDs = ds.filter(num -> num > 10);

        SingleOutputStreamOperator<Tuple2<Integer, Integer>> noRebalanceMap = filterDs.map(new RichMapFunction<Long, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> map(Long aLong) throws Exception {
                return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(), 1);
            }
        }).keyBy(t->t.f0).sum(1);

        SingleOutputStreamOperator<Tuple2<Integer, Integer>> rebalanceMap = filterDs.rebalance().map(new RichMapFunction<Long, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> map(Long aLong) throws Exception {
                return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(), 1);
            }
        }).keyBy(t -> t.f0).sum(1);


        //TODO 4.sink操作
        noRebalanceMap.print("noRebalanceMap");
        rebalanceMap.print("rebalanceMap");

        //TODO 5.执行程序
        env.execute();
    }
}
